package com.navinfo.platform.flink.cdc.util;

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.flink.api.java.utils.ParameterTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.SQLException;

/**
 * DBUtils工具类
 * 主要用来初始化数据库连接池,并能获取Connection对象
 */
public class DBUtils {
    private static final Logger logger = LoggerFactory.getLogger(DBUtils.class);

    /*
     * 这个dataSource,在每个slot上,是唯一的
     */
    private static HikariDataSource dataSource;

    /**
     * 初始化数据库连接池,此方法只被调用一次
     * 在IDE中执行时,此方法只会被执行一次
     * 预期在多个TaskManager中的多个slot中执行时,此方法会被执行多次,在每个slot上创建一个连接池
     */
    public static synchronized void init(ParameterTool config) {
        if (dataSource != null) {
            return;
        }
        logger.info("开始初始化数据库Hikari连接池");
        HikariConfig hikariConfig = new HikariConfig();
        hikariConfig.setJdbcUrl(config.get("sink.jdbc.url"));
        hikariConfig.setUsername(config.get("sink.jdbc.username"));
        hikariConfig.setPassword(config.get("sink.jdbc.password"));
        hikariConfig.addDataSourceProperty("maximumPoolSize", "10");
        hikariConfig.addDataSourceProperty("connectionTimeout", "3000");
        hikariConfig.addDataSourceProperty("cachePrepStmts", "true");
        hikariConfig.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
        dataSource = new HikariDataSource(hikariConfig);
        logger.info("数据库Hikari连接池创建成功");
    }

    /**
     * 从连接池中获取一个连接
     */
    public static Connection getConnection() throws SQLException {
        return dataSource.getConnection();
    }
}
